package controller;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import pojo.UserBehavior;

import java.net.URL;

/*
* 读取埋点日志，统计网站平台总浏览量PV 2021-05-22完成
*
* */
public class PageViewController {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        URL resource = PageViewController.class.getResource("/UserBehavior.csv");
        DataStream<String> fileDataStream = streamEnv.readTextFile(resource.getPath());

        DataStream<UserBehavior> watermarkDataStream = fileDataStream.map(line -> {
            String[] fields = line.split(",");
            return new UserBehavior(Long.valueOf(fields[0]), Long.valueOf(fields[1]), Integer.valueOf(fields[2]), fields[3], Long.valueOf(fields[4]));
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserBehavior>noWatermarks()
                        .withTimestampAssigner((SerializableTimestampAssigner<UserBehavior>) (element, recordTimestamp) -> element.getTimestamp() * 1000L)
        );

        DataStream<Tuple2<String, Integer>> pvResult = watermarkDataStream.filter(line -> "pv".equals(line.getBehavior()))
                .map((MapFunction<UserBehavior, Tuple2<String, Integer>>) value -> new Tuple2<>("pv",1))
                .returns(TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(String.class,Integer.class))
                .keyBy(value -> value.f0)
                .window(TumblingEventTimeWindows.of(Time.hours(1)))
                .sum(1);

        pvResult.print();

        streamEnv.execute();
    }
}
